package com.shxzz.easy12306.services.ticket.mq.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.text.CharSequenceUtil;
import com.shxzz.easy12306.framework.starter.common.mq.constant.OrderRocketMQConstant;
import com.shxzz.easy12306.framework.starter.common.mq.event.CanalBinlogEvent;
import com.shxzz.easy12306.framework.starter.designpattern.strategy.AbstractStrategyChoose;
import com.shxzz.easy12306.services.ticket.common.constant.RocketMQConstant;
import com.shxzz.easy12306.services.ticket.common.enums.CanalExecuteStrategyMark;
import com.shxzz.easy12306.services.ticket.config.TicketRemainCacheUpdateProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import static com.shxzz.easy12306.services.ticket.config.TicketRemainCacheUpdateProperties.BINLOG_UPDATE_TYPE;

/**
 * canal binlog 消费者
 *
 * @author shxzz
 * @create 2024-03-18-23:07
 */
@Slf4j
@RequiredArgsConstructor
@Component
@RocketMQMessageListener(
        topic = OrderRocketMQConstant.CANAL_BINLOG_TOPIC,
        consumerGroup = RocketMQConstant.CANAL_BINLOG_CG
)
public class CanalBinlogConsumer implements RocketMQListener<CanalBinlogEvent> {

    private final AbstractStrategyChoose abstractStrategyChoose;
    private final TicketRemainCacheUpdateProperties ticketRemainCacheUpdateProperties;
    @Override
    public void onMessage(CanalBinlogEvent message) {
        if (Boolean.TRUE.equals(message.getIsDdl())
                || CollUtil.isEmpty(message.getOld())
                || !CharSequenceUtil.equals(message.getType(), "UPDATE")) {
            return;
        }
        if(!CharSequenceUtil.equals(ticketRemainCacheUpdateProperties.getTicketUpdateType(), BINLOG_UPDATE_TYPE)) return;
        log.info("来自表 [{}] 的 binlog 更新", message.getDatabase() + ":" + message.getTable());
        abstractStrategyChoose.chooseAndExecute(
                message.getTable(),
                message,
                CanalExecuteStrategyMark.isPatternMatch(message.getTable())
        );
    }
}
